package com.itttq.im.ws.netty;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.itttq.im.common.constant.Constant;
import com.itttq.im.common.constant.RedisConstant;
import com.itttq.im.conf.WsProperties;
import com.itttq.im.model.to.MsgTo;
import com.itttq.im.service.MsgService;
import com.itttq.im.utils.RedisUtil;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.*;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;

/**
 * 消息处理类。TODO 后面探索直接传ProtoBuf对象
 *
 * @Author GXG
 */
@Slf4j
@Component
public class MessageHandler {
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
    private static final String WS_SERVER = "http://%s/chat/sendMsg";
    @Autowired
    private MsgService msgService;
    @Autowired
    private RestTemplate restTemplate;
    @Autowired
    private RedisUtil<String> stringRedisUtil;

    /**
     * 处理字符串格式的消息，含处理客户端心跳信息
     *
     * @param msgToStr
     */
    public void handleMsg(Channel currentChannel, String msgToStr) {
        try {
            handleMsg(currentChannel, OBJECT_MAPPER.readValue(msgToStr, MsgTo.class));
        } catch (IOException e) {
            log.error("发送消息JSON解析异常，msg={}", msgToStr, e);
        }
    }

    /**
     * 处理消息，不处理心跳消息
     *
     * @param msgTo
     */
    public void handleMsg(MsgTo msgTo) {
        switch (MsgTo.MsgType.valueOf(msgTo.getType())) {
            case NOTICE:
            case P2P:
                // 根据发送目的获取channel
                sendMsg(msgTo);
                // 保存
                msgService.save(msgTo);
                break;
            case GROUP:
                // 从Redis获取群组用户信息并发送（不含自己）
                sendGroupMsg(msgTo);
                msgService.save(msgTo);
                break;
            case BROADCAST:
                // 发送给所有的用户（不含自己）
                sendGroupMsg(msgTo);
                break;
            default:
        }
    }

    /**
     * 处理消息，心跳在此方法处理；非心跳消息在其他方法处理
     *
     * @param msgTo
     */
    public void handleMsg(Channel currentChannel, MsgTo msgTo) {
        // TODO 心跳后续改为前端 Ping/Pong方式实现
        if (msgTo.getType().equals(MsgTo.MsgType.HEARTBEAT.getCode())) {
            // 如果是客户端对心跳的响应消息，则直接舍弃调
            if (Constant.CLIENT.equals(msgTo.getBody())) {
                // 前端的心跳消息。则直接响应一个OK
                msgTo.setBody(Constant.HEARTBEAT_RESPONSE);
                sendMsg(currentChannel, msgTo);
            }
        } else {
            handleMsg(msgTo);
        }
    }

    /**
     * 单发消息
     *
     * @param msgTo
     */
    public void sendMsg(MsgTo msgTo) {
        Channel channel = WsServerHandler.userChannel.get(msgTo.getTo());
        if (null == channel) {
            try {
                log.info("转发发送消息：{}", msgTo);
                forwardMsg(msgTo);
            } catch (JsonProcessingException e) {
                log.error("转发消息异常", e);
            }
        } else {
            log.info("正常发送消息：{}", msgTo);
            sendMsg(channel, msgTo);
        }
    }

    /**
     * 单发消息
     *
     * @param msgTo
     */
    public static void sendMsg(Channel channel, MsgTo msgTo) {
        try {
            if (null == channel) {
                log.error("频道不存在");
                return;
            }
            if (!channel.isWritable()) {
                log.error("频道【user={}，group={}】不可用", channel.attr(Attributes.USER_ID).get(), channel.attr(Attributes.GROUP_ID).get());
                return;
            }
            channel.writeAndFlush(getTextWsFrame(msgTo));
        } catch (JsonProcessingException e) {
            log.error("发送消息JSON解析异常，msg={}", msgTo, e);
        }
    }

    /**
     * 发送群消息,此时所有服务器的用户都能收到群消息
     */
    public void sendGroupMsg(MsgTo msgTo) {
        try {
            // 获取发送用户对象，TODO GXG 同时发送给其他的服务器
            if (MsgTo.MsgType.BROADCAST.getCode() == msgTo.getType().intValue()) {
                sendMsgBatch(WsServerHandler.channelGroup, msgTo);
            } else if (MsgTo.MsgType.GROUP.getCode() == msgTo.getType().intValue()) {
                sendMsgBatch(WsServerHandler.groupChannels.get(msgTo.getTo()), msgTo);
            }
        } catch (JsonProcessingException e) {
            log.error("JSON解析异常", e);
        }
    }

    /**
     * 批量发送消息
     */
    public static void sendMsgBatch(ChannelGroup channelGroup, MsgTo msg) throws JsonProcessingException {
        channelGroup.writeAndFlush(getTextWsFrame(msg));
        // TODO 发送到其他服务器上
//        forwardMsg();
    }

    /**
     * 转换为JSON格式，抛出异常，由业务自己处理
     *
     * @param msg
     * @return
     */
    public static TextWebSocketFrame getTextWsFrame(MsgTo msg) throws JsonProcessingException {
        return new TextWebSocketFrame(OBJECT_MAPPER.writeValueAsString(msg));
    }

    @Async
    public void forwardMsg(MsgTo msgTo) throws JsonProcessingException {
        // 如果不存在，则从Redis缓存查看此用户所在服务器，向所在服务器发送消息
        String host = stringRedisUtil.getMapValue(RedisConstant.getKeyUserServer(), msgTo.getTo());
        // 为空或本地地址则不再转发，避免因为缓存数据存在，本地不存在的异常情况，导致死循环发送
        if (StringUtils.isEmpty(host) || host.startsWith(WsProperties.localIp + Constant.COLON + WsProperties.httpPort)) {
            log.info("HOST异常导致转发消息失败：host={}，msg={}", host, msgTo);
            return;
        }
        forwardMsg(msgTo, host);
    }

    private void forwardMsg(MsgTo msgTo, String host) throws JsonProcessingException {
        String url = String.format(WS_SERVER, host);
        log.info("转发消息：add={}，msg={}", url, msgTo);
        HttpHeaders headers = new HttpHeaders();
        //这里设置的是以payLoad方式提交数据，对于Payload方式，提交的内容一定要是String，且Header要设为“application/json”
        headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
        HttpEntity<String> requestEntity = new HttpEntity<>(OBJECT_MAPPER.writeValueAsString(msgTo), headers);
        ResponseEntity<String> response = restTemplate.postForEntity(url, requestEntity, String.class);
        if (HttpStatus.OK.equals(response.getStatusCode())) {
            log.info("消息转发成功，地址={}，参数={}", host, requestEntity.getBody());
        } else {
            log.info("消息转发失败，地址={}，参数={}", host, requestEntity.getBody());
        }
    }
}
